package com.ververica.cdc.guass.source.kafka.table;

import com.ververica.cdc.guass.Constants;
import com.ververica.cdc.guass.source.kafka.GaussKafkaSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.connector.source.SourceFunctionProvider;
import org.apache.flink.types.RowKind;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;

public class GaussKafkaDynamicTableSource implements ScanTableSource {

    private final Logger log = LoggerFactory.getLogger(this.getClass());

    private final List<PhysicalColumn> physicalColumns;
    private final String topic;
    private final String bootstrapServers;
    private final String groupId;
    private final String tableName;
    private final String scanStartupMode;

    public GaussKafkaDynamicTableSource(
            List<PhysicalColumn> physicalColumns,
            String topic,
            String bootstrapServers,
            String groupId,
            String tableName,
            String scanStartupMode) {
        this.physicalColumns = physicalColumns;
        this.topic = topic;
        this.bootstrapServers = bootstrapServers;
        this.groupId = groupId;
        this.tableName = tableName;
        this.scanStartupMode = scanStartupMode;
    }

    @Override
    public ChangelogMode getChangelogMode() {
        return ChangelogMode.newBuilder()
                .addContainedKind(RowKind.INSERT)
                .addContainedKind(RowKind.UPDATE_AFTER)
                .addContainedKind(RowKind.DELETE)
                .build();
    }

    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {

        final SourceFunction sourceFunction = createSourceFunction();

        // Use parallelism that matches Kafka partition count
        return SourceFunctionProvider.of(sourceFunction, true);
    }

    private ChangeEventToRowDataSourceFunction createSourceFunction() {
        log.info("Creating source function for table: {}", tableName);
        GaussKafkaSourceFunction gaussKafkaSourceFunction = new GaussKafkaSourceFunction(
                topic, bootstrapServers, groupId, tableName, scanStartupMode);
        return new ChangeEventToRowDataSourceFunction(gaussKafkaSourceFunction, physicalColumns);
    }

    @Override
    public DynamicTableSource copy() {
        return new GaussKafkaDynamicTableSource(
                physicalColumns,
                topic,
                bootstrapServers,
                groupId,
                tableName,
                scanStartupMode
        );
    }

    @Override
    public String asSummaryString() {
        return Constants.GAUSS_KAFKA_CDC_CONNECTOR_NAME;
    }
}